package com.yanggu.flink.datastream_api.partition

import org.apache.flink.api.common.functions.Partitioner
import org.apache.flink.streaming.api.scala._

/**
 * 当Flink提供的所有分区策略都不能满足用户的需求时，我们可以通过使用partitionCustom()方法来自定义分区策略
 */
//CustomPartition:1> 2
//CustomPartition:2> 1
object CustomPartitionDemo {

  def main(args: Array[String]): Unit = {
    // 创建执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    // 将自然数按照奇偶分区// 将自然数按照奇偶分区

    env
      .fromElements(1, 2, 3, 4, 5, 6, 7, 8)
      .partitionCustom(new Partitioner[Int]() {
        def partition(key: Int, numPartitions: Int) = key % 2
      }, data => data)
      .print("CustomPartition").setParallelism(2)

    env.execute()
  }

}
